package com.weishao.learn;

import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.KettleLogStore;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepIOMetaInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.mergerows.MergeRowsMeta;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;
import org.pentaho.di.trans.steps.tableoutput.TableOutputMeta;

/**
 * 基于Kettle计算同一张表数据前后两次的变化量
 * 
 * 原理：使用kettle的"合并记录"组件MergeRowsMeta
 * 
 * 说明：该步骤用于将两个不同来源的数据合并，这两个来源的数据分别为旧数据和新数据，该步骤将旧数据和新数据按
 * 照指定的关键字匹配、比较、合并。
 * 
 * 旧数据来源：旧数据来源的步骤
 * 新数据来源。新数据来源的步骤
 * 
 * 标志字段：设置标志字段的名称，标志字段用于保存比较的结果，比较结果有下列几种。
 * 1. “identical” – 旧数据和新数据一样
 * 2. “changed” – 数据发生了变化;
 * 3. “new” – 新数据中有而旧数据中没有的记录
 * 4. “deleted” –旧数据中有而新数据中没有的记录
 * 
 * 关键字段：用于定位两个数据源中的同一条记录。
 * 比较字段：对于两个数据源中的同一条记录中，指定需要比较的字段。
 * 
 * 合并后的数据将包括旧数据来源和新数据来源里的所有数据，对于变化的数据，使用新数据代替旧数据，同时在结果里用一
 * 个标示字段，来指定新旧数据的比较结果。注意：
 * 1、旧数据和新数据需要事先按照关键字段排序。
 * 2、旧数据和新数据要有相同的字段名称。
 * 
 * @author tang
 *
 * 参考内容：
 * 1、http://blog.itpub.net/14366449/viewspace-1005646/
 * 2、https://blog.csdn.net/weixin_33698823/article/details/86123874
 * 
 */
public class MergeRowApplication {

	public static void main(String[] args) throws KettleException {
		//旧表的查询SQL
		String oldQuerySql="SELECT DEPTNO, NAME  FROM test_for_tang.DEPT_1";
		//新表的查询SQL
		String newQuerySql="SELECT DEPTNO, NAME  FROM test_for_tang.DEPT_2";
		//比较的字段key，通常为表的主键字段
		String fieldsPrimaryKey[]= {"DEPTNO"};
		//比较的字段value，通常为表中除主键外的其他字段
		String fieldsCompareValue[]= {"NAME"};
		//计算结果存储的库schema名称
		String diffSchemaName="test_for_tang";
		//计算结果存储的库table名称
		String diffTableName="DEPT_DIFF";
		//计算结果中标记字段的名称，标记字段的取值为：identical-未变；new-新增；changed-更新；deleted-删除；
		String flagFieldName="diff";
		
		KettleEnvironment.init();
		
		TransMeta transMeta = new TransMeta();
		transMeta.setName("trans");
		
		DatabaseMeta db = new DatabaseMeta("source", "mysql", "Native(JDBC)", "172.17.207.210", "test_for_tang",	"3306", "tangyibo", "tangyibo");
		db.addExtraOption(db.getPluginId(), "characterEncoding", "utf8"); 
		db.setConnectSQL("set names utf8;");
		transMeta.addDatabase(db);

		TableInputMeta oldTableInput = new TableInputMeta();
		oldTableInput.setDatabaseMeta(db);
		
		oldTableInput.setSQL(oldQuerySql);
		StepMeta oldTableInputMetaStep = new StepMeta("old",oldTableInput);
		transMeta.addStep(oldTableInputMetaStep);
	
		TableInputMeta newTableInput = new TableInputMeta();
		newTableInput.setDatabaseMeta(db);
		newTableInput.setSQL(newQuerySql);
		StepMeta newTableInputMetaStep = new StepMeta("new",newTableInput);
		transMeta.addStep(newTableInputMetaStep);

		MergeRowsMeta mergeRowsMeta = new MergeRowsMeta();
		StepIOMetaInterface stepIOMeta = mergeRowsMeta.getStepIOMeta();
		stepIOMeta.getInfoStreams().get(0).setStepMeta(oldTableInputMetaStep);
		stepIOMeta.getInfoStreams().get(1).setStepMeta(newTableInputMetaStep);
		mergeRowsMeta.setFlagField(flagFieldName);
		mergeRowsMeta.setKeyFields(fieldsPrimaryKey);
		mergeRowsMeta.setValueFields(fieldsCompareValue);
		
		StepMeta mergeStepMeta = new StepMeta("recordMerge",mergeRowsMeta);
		transMeta.addStep(mergeStepMeta);

		transMeta.addTransHop(new TransHopMeta(oldTableInputMetaStep,mergeStepMeta));
		transMeta.addTransHop(new TransHopMeta(newTableInputMetaStep,mergeStepMeta));

		TableOutputMeta t_output = new TableOutputMeta();
		t_output.setDatabaseMeta(db);
		t_output.setSchemaName(diffSchemaName);
		t_output.setTableName(diffTableName);
		t_output.setCommitSize(10000);	
		t_output.setTruncateTable(true);
		StepMeta output = new StepMeta("tableOutput", t_output);
		transMeta.addStep(output);
		transMeta.addTransHop(new TransHopMeta(mergeStepMeta, output));
		
		//打印转换的XML配置
		String transXml = transMeta.getXML();
		System.out.println("=================");
		System.out.println(transXml);
		
		Trans trans = new Trans(transMeta);
		try {
			trans.execute(null);
			System.out.println("start............");
			trans.waitUntilFinished();
			System.out.println("end............");
			if (trans.getErrors() > 0) {
				String errMsg = KettleLogStore.getAppender().getBuffer(trans.getLogChannelId(), false).toString();
				throw new KettleException(errMsg);
			}
		} finally {
			trans.cleanup();
		}
	}

}
